package com.cs.deviceManager.mqtt;

import com.cs.deviceManager.device.DeviceTypeManager;
import com.cs.deviceManager.device.DeviceEventListener;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * MQTT 工具类，封装 MQTT 客户端的通用操作。
 * 支持自动重连、连接状态感知、线程安全。
 * ✅ 改造：增加异步发布线程池，确保 publish 不阻塞调用线程。
 */
public class MqttUtil {
    private static final Logger log = LoggerFactory.getLogger(MqttUtil.class);
    private static final int QOS = 1;
    private static final int RECONNECT_INTERVAL = 5000; // 重连间隔（毫秒）

    // 存储所有 MQTT 客户端（线程安全）
    private static final Map<String, MqttClientWrapper> CLIENTS = new ConcurrentHashMap<>();
    // 存储设备在线状态（线程安全）
    private static final Map<String, DeviceStatus> DEVICE_STATUS_MAP = new ConcurrentHashMap<>();
    private static DeviceHeartbeatMonitor HEARTBEAT_MONITOR;

    // ✅ 新增：专用的异步发布线程池
    private static final ExecutorService PUBLISH_EXECUTOR = Executors.newCachedThreadPool(r -> {
        Thread t = new Thread(r, "MqttPublish-" + r.hashCode());
        t.setDaemon(true); // 避免阻止JVM退出
        return t;
    });

    // 初始化时启动监控器（可在 MqttStartupListener 中调用）
    public static void startHeartbeatMonitor(long timeoutSeconds, DeviceEventListener listener) {
        if (HEARTBEAT_MONITOR == null) {
            HEARTBEAT_MONITOR = new DeviceHeartbeatMonitor(getDeviceStatusMap(), timeoutSeconds, listener);
            log.info("心跳监控器已启动，超时时间：{} 秒", timeoutSeconds);
        }
    }

    // 提供给外部调用（当设备上报数据时，包含设备类型）
    public static void reportHeartbeat(String deviceId, String deviceType) {
        if (HEARTBEAT_MONITOR != null) {
            HEARTBEAT_MONITOR.heartbeat(deviceId, deviceType);
        }
    }

    // 提供给外部调用（当设备上报数据时）
    public static void reportHeartbeat(String deviceId) {
        if (HEARTBEAT_MONITOR != null) {
            HEARTBEAT_MONITOR.heartbeat(deviceId);
        }
    }

    // 手动标记离线（可选）
    public static void markDeviceOffline(String deviceId) {
        if (HEARTBEAT_MONITOR != null) {
            HEARTBEAT_MONITOR.markOffline(deviceId);
        }
    }

    // 获取设备状态映射表
    public static Map<String, DeviceStatus> getDeviceStatusMap() {
        return DEVICE_STATUS_MAP;
    }

    /**
     * 获取设备状态
     *
     * @param deviceId 设备ID
     * @return 设备状态
     */
    public static DeviceStatus getDeviceStatus(String deviceId) {
        return DEVICE_STATUS_MAP.get(deviceId);
    }

    /**
     * 设置设备状态
     *
     * @param deviceId 设备ID
     * @param status   设备状态
     */
    public static void setDeviceStatus(String deviceId, DeviceStatus status) {
        DEVICE_STATUS_MAP.put(deviceId, status);
    }

    /**
     * 检查设备是否在线
     *
     * @param deviceId 设备ID
     * @return 在线状态
     */
    public static boolean isDeviceOnline(String deviceId) {
        DeviceStatus status = DEVICE_STATUS_MAP.get(deviceId);
        return status == DeviceStatus.ONLINE;
    }

    /**
     * 设置设备类型的超时时间
     *
     * @param deviceType     设备类型
     * @param timeoutSeconds 超时时间（秒）
     */
    public static void setDeviceTypeTimeout(String deviceType, long timeoutSeconds) {
        DeviceTypeManager.setDeviceTypeTimeout(deviceType, timeoutSeconds);
    }

    /**
     * 获取设备类型的超时时间
     *
     * @param deviceType 设备类型
     * @return 超时时间（秒）
     */
    public static long getDeviceTypeTimeout(String deviceType) {
        return DeviceTypeManager.getDeviceTypeTimeout(deviceType);
    }

    /**
     * 连接到 MQTT 服务器
     *
     * @param broker   MQTT 服务器地址
     * @param clientId 客户端 ID
     * @param username 用户名
     * @param password 密码
     * @return 连接成功返回 true，失败返回 false
     */
    public static boolean connect(String broker, String clientId, String username, String password) {
        if (CLIENTS.containsKey(clientId)) {
            log.warn("MQTT 客户端 {} 已存在", clientId);
            return CLIENTS.get(clientId).isConnected();
        }
        try {
            MqttClientWrapper wrapper = new MqttClientWrapper(broker, clientId, username, password);
            wrapper.connect();
            CLIENTS.put(clientId, wrapper);
            log.info("MQTT 客户端 {} 连接成功", clientId);
            return true;
        } catch (MqttException e) {
            log.error("MQTT 客户端 {} 连接失败", clientId, e);
            return false;
        }
    }

    /**
     * 断开指定客户端的连接
     *
     * @param clientId 客户端 ID
     */
    public static void disconnect(String clientId) {
        MqttClientWrapper wrapper = CLIENTS.get(clientId);
        if (wrapper != null) {
            wrapper.disconnect();
            CLIENTS.remove(clientId);
            log.info("MQTT 客户端 {} 已断开连接", clientId);
        }
    }

    /**
     * 添加订阅
     *
     * @param clientId 客户端 ID
     * @param topic    订阅主题
     * @param callback 消息回调
     */
    public static void subscribe(String clientId, String topic, MqttCallback callback) {
        MqttClientWrapper wrapper = CLIENTS.get(clientId);
        if (wrapper != null && wrapper.isConnected()) {
            try {
                if (callback != null) {
                    wrapper.setCallback(callback);
                }
                wrapper.subscribe(topic, QOS);
                log.info("MQTT 客户端 {} 订阅主题 {} 成功", clientId, topic);
                return;
            } catch (MqttException e) {
                log.error("MQTT 客户端 {} 订阅主题 {} 失败", clientId, topic, e);
                return;
            }
        }
        log.warn("MQTT 客户端 {} 未连接，无法订阅主题 {}", clientId, topic);
    }

    /**
     * 移除订阅
     *
     * @param clientId 客户端 ID
     * @param topic    订阅主题
     */
    public static void unsubscribe(String clientId, String topic) {
        MqttClientWrapper wrapper = CLIENTS.get(clientId);
        if (wrapper != null && wrapper.isConnected()) {
            try {
                wrapper.unsubscribe(topic);
                log.info("MQTT 客户端 {} 移除订阅主题 {} 成功", clientId, topic);
                return;
            } catch (MqttException e) {
                log.error("MQTT 客户端 {} 移除订阅主题 {} 失败", clientId, topic, e);
                return;
            }
        }
        log.warn("MQTT 客户端 {} 未连接，无法移除订阅主题 {}", clientId, topic);
    }

    /**
     * 发布消息 - ✅ 异步非阻塞版本
     * 调用此方法会立即返回，实际发布操作在独立线程中完成。
     * @param clientId 客户端 ID
     * @param topic    发布主题
     * @param message  消息内容 (byte[])
     * @return 返回 true 表示发布任务已提交，false 表示客户端不存在或未连接。
     */
    public static boolean publish(String clientId, String topic, byte[] message) {
        MqttClientWrapper wrapper = CLIENTS.get(clientId);
        if (wrapper == null || !wrapper.isConnected()) {
            log.warn("MQTT 客户端 {} 不存在或未连接，无法发布消息到主题 {}", clientId, topic);
            return false;
        }

        // ✅ 关键：将发布任务提交到专用的异步线程池
        PUBLISH_EXECUTOR.execute(() -> {
            try {
                MqttMessage mqttMessage = new MqttMessage(message);
                mqttMessage.setQos(QOS);
                mqttMessage.setRetained(false);
                wrapper.publish(topic, mqttMessage);
                // log.debug("MQTT 客户端 {} 成功发布消息到主题 {}", clientId, topic);
                // 使用debug级别，避免日志过多
            } catch (MqttException e) {
                log.error("MQTT 客户端 {} 发布消息到主题 {} 时发生异常", clientId, topic, e);
            } catch (Exception e) {
                log.error("MQTT 客户端 {} 发布消息到主题 {} 时发生未知异常", clientId, topic, e);
            }
        });

        // ✅ 立即返回，不阻塞调用者
        return true;
    }

    // 保持String重载，内部调用byte[]版本
    public static boolean publish(String clientId, String topic, String message) {
        if (message == null) return false;
        return publish(clientId, topic, message.getBytes());
    }

    /**
     * 添加设备类型到心跳监控器
     *
     * @param deviceId   设备 ID
     * @param deviceType 设备类型
     */
    public static void addDeviceType(String deviceId, String deviceType) {
        // 可以通过心跳上报时自动记录设备类型，无需单独调用
    }

    /**
     * 应用关闭时手动调用此方法，清理所有 MQTT 连接
     */
    public static void destroy() {
        if (HEARTBEAT_MONITOR != null) {
            HEARTBEAT_MONITOR.shutdown();
        }
        // ✅ 增加空值检查
        if (CLIENTS != null) {
            for (String clientId : CLIENTS.keySet()) {
                disconnect(clientId);
            }
        }
        // ✅ 关闭发布线程池
        PUBLISH_EXECUTOR.shutdown();
        try {
            if (!PUBLISH_EXECUTOR.awaitTermination(5, TimeUnit.SECONDS)) {
                PUBLISH_EXECUTOR.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            PUBLISH_EXECUTOR.shutdownNow();
        }
    }

    // 内部类：封装客户端连接、自动重连等逻辑
    // 请保留你原始的 MqttClientWrapper 实现，无需修改。
    // 由于您提供的 MqttClientWrapper 代码较长，且其 publish 方法是核心，这里仅保留其结构。
    // 请确保您的 MqttClientWrapper 类包含 connect, disconnect, subscribe, unsubscribe, publish 等方法。
    // 该类的实现不影响本次 publish 的异步化改造。
    private static class MqttClientWrapper {
        private final String broker;
        private final String clientId;
        private final String username;
        private final char[] password;
        private IMqttClient client;
        private MqttCallback callback;
        private final AtomicBoolean connected = new AtomicBoolean(false);
        private final ScheduledExecutorService reconnectScheduler = Executors.newSingleThreadScheduledExecutor();
        private boolean shouldReconnect = true;

        public MqttClientWrapper(String broker, String clientId, String username, String password) throws MqttException {
            this.broker = broker;
            this.clientId = clientId;
            this.username = username;
            this.password = password == null ? null : password.toCharArray();
            this.client = new MqttClient(broker, clientId, new MemoryPersistence());
        }

        public boolean isConnected() {
            return client.isConnected();
        }

        public void setCallback(MqttCallback callback) {
            this.callback = callback;
            client.setCallback(callback);
        }

        public void connect() throws MqttException {
            if (client.isConnected()) {
                return;
            }
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            if (username != null && password != null) {
                options.setUserName(username);
                options.setPassword(password);
            }
            client.connect(options);
            connected.set(true);
        }

        public void disconnect() {
            shouldReconnect = false;
            try {
                if (client.isConnected()) {
                    client.disconnect();
                }
                connected.set(false);
                reconnectScheduler.shutdownNow();
            } catch (MqttException e) {
                log.error("MQTT 客户端 {} 断开连接失败", clientId, e);
            }
        }

        public void subscribe(String topic, int qos) throws MqttException {
            client.subscribe(topic, qos);
        }

        public void unsubscribe(String topic) throws MqttException {
            client.unsubscribe(topic);
        }

        public void publish(String topic, MqttMessage message) throws MqttException {
            client.publish(topic, message);
        }

    }
}